package com.boarsoft.boar.sequence.biz.impl;

import java.util.List;

import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.boarsoft.boar.sequence.biz.SequenceSegmentBiz;
import com.boarsoft.boar.sequence.entity.SequenceInfo;
import com.boarsoft.boar.sequence.entity.SequenceSegment;
import com.boarsoft.common.Util;
import com.boarsoft.common.dao.PagedResult;
import com.boarsoft.hibernate.biz.SimpleBizImpl;

@Component("sequenceSegmentBiz")
public class SequenceSegmentBizImpl extends SimpleBizImpl implements SequenceSegmentBiz {

	@Override
	@Transactional
	public void save(SequenceSegment o) {
		if (Util.strIsEmpty(o.getId())) {
			dao.save(o);
		} else {
			dao.update(o);
		}
	}

	@Override
	@Transactional(readOnly = true)
	public SequenceSegment get(String id) {
		return dao.get(SequenceSegment.class, id);
	}

	@Override
	@Transactional
	public void delete(String id) {
		dao.delete(SequenceSegment.class, id);
	}

	@Override
	@Transactional(readOnly = true)
	public PagedResult<SequenceSegment> list(String sequenceId, String orderBy, int pageNo, int pageSize) {
		StringBuilder sb = new StringBuilder();
		if (Util.strIsNotEmpty(sequenceId)) {
			sb.append(" and sequenceId='").append(sequenceId).append("'");
		}
		int total = dao.getTotal(SequenceSegment.class, sb.toString());
		if (Util.strIsNotEmpty(orderBy)) {
			sb.append(" order by ").append(orderBy);
		}
		List<SequenceSegment> lt = dao.list(SequenceSegment.class, sb.toString());
		return new PagedResult<SequenceSegment>(total, lt, pageNo, pageSize);
	}

	@Override
	@Transactional
	public SequenceSegment assign(SequenceInfo seq, String node) {
		// 根据node（远程节点IP，或其它唯一标识）到数据库查询（未用完的）号段
		SequenceSegment seg = this.getCurrent(seq.getId(), node);
		// 如果有已分配的号段，且未用完，就返回
		if (seg != null) {
			return seg;
		}
		// 乐观锁实现
		while (true) {
			// 取当前号段的下一个号段
			long begin = seq.getBeginNo();
			// 取当前号段（上一次分配的号段）
			SequenceSegment curr = this.find(seq.getId(), seq.getLastNo());
			if (curr != null) {
				begin = curr.getEndNo() + 1;
			}
			long end = begin + seq.getSegLen() - 1;
			// 判断号段是否越界
			if (end > seq.getEndNo()) {
				// 如果号段超出序列的最大值，修改当前号段为第一个号段
				if (seq.isRecyclable()) {
					begin = seq.getBeginNo();
				} else {
					return null;
				}
			}
			// 尝试锁定该号段（乐观锁）
			StringBuilder sb = new StringBuilder();
			sb.append("update ").append(SequenceInfo.class.getName());
			sb.append(" set lastNo = ").append(begin);
			sb.append(" where id = '").append(seq.getId());
			sb.append("' and lastNo = ").append(seq.getLastNo());
			switch (dao.executeUpdate(sb.toString())) {
			case 1: // 如果更新sequence_info成功，表示移动分段成功
				break;
			case 0: // 如果current已经被别人修改，更新失败，继续重试
				continue;
			default:// Impossible case
				throw new IllegalStateException(String.format(//
						"Duplicated sequence info %s/%d", seq.getId(), seq.getLastNo()));
			}
			// 判断这个号段是否第一次创建
			seg = this.find(seq.getId(), begin);
			if (seg == null) {
				// 创建新号段（如果已存在就更新，不存在就创建）
				seg = new SequenceSegment();
				seg.setSequenceId(seq.getId());
				seg.setBeginNo(begin);
				seg.setEndNo(end);
				seg.setNode(node);
				dao.save(seg);
			} else {
				seg.setNode(node);
				dao.update(seg);
			}
			return seg;
		}
	}

	/**
	 * 查找指定号段
	 * 
	 * @param seqId
	 * @param begin
	 * @param end
	 * @return
	 */
	@Transactional(readOnly = true)
	protected SequenceSegment find(String seqId, long begin) {
		StringBuilder sb = new StringBuilder();
		sb.append("and sequenceId = '").append(seqId);
		sb.append("' and beginNo = ").append(begin);
		// sb.append(" and lastNo = endNo");
		return dao.find(SequenceSegment.class, sb.toString());
	}

	/**
	 * 查询已分配给指定节点的未用完的号段
	 * 
	 * @param seqId
	 * @param node
	 * @return
	 */
	@Transactional(readOnly = true)
	protected SequenceSegment getCurrent(String seqId, String node) {
		StringBuilder sb = new StringBuilder();
		sb.append("and sequenceId = '").append(seqId);
		sb.append("' and node = '").append(node);
		sb.append("' and lastNo < endNo");
		return dao.find(SequenceSegment.class, sb.toString());
	}
}
